package org.example.common.configure.kafka;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;

import java.util.Map;

/**
 * 描述: 启用 kafka
 *
 * @author yanpenglei
 * @create 2017-10-16 18:25
 **/

/**
 * @author tianshl
 * @version 2017/9/1 下午04:07
 */
@Configuration
@EnableKafka
public class KafkaConfiguration {

    /**
     * 从 Spring Boot 默认的 KafkaProperties 构建 ConsumerFactory，避免手工硬编码属性。
     * 使用 application.yml 中的 spring.kafka.consumer.* 配置。
     */
    @Bean
    public ConsumerFactory<String, String> consumerFactory(KafkaProperties kafkaProperties) {
        Map<String, Object> consumerProps = kafkaProperties.buildConsumerProperties();
        // 推荐将 groupId 放在 @KafkaListener 上配置，而不在这里硬编码
        // 可根据需要覆盖部分默认属性：
        // consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        return new DefaultKafkaConsumerFactory<>(consumerProps);
    }

    /**
     * 全局复用的 Listener 容器工厂：
     * - 是否批量消费、并发度、Ack 模式由配置控制，保持灵活性与类型安全。
     * - 具体的 groupId 在各自的 @KafkaListener 上通过 groupId / properties 指定。
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory,
            @Value("${spring.kafka.listener.concurrency:3}") Integer concurrency,
            @Value("${spring.kafka.listener.type:batch}") String listenerType,
            @Value("${spring.kafka.listener.ack-mode:MANUAL_IMMEDIATE}") String ackMode
    ) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(concurrency);
        // 批量 or 单条
        factory.setBatchListener("batch".equalsIgnoreCase(listenerType));
        // Ack 模式
        AbstractMessageListenerContainer.AckMode mode = AbstractMessageListenerContainer.AckMode.valueOf(ackMode);
        factory.getContainerProperties().setAckMode(mode);
        return factory;
    }
}
